package bigdata.jobclean.test

import bigdata.jobclean.Job
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{Encoder, Encoders, SparkSession}

object Analysis {

  /*
  统计下清理结果
   */
  val jobEncoder: Encoder[Job] = Encoders.product[Job]
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .master("local[2]").appName("jobClean").getOrCreate()
    //分组检查下数据
    spark.sparkContext.setLogLevel("WARN")
    val schema = StructType(Array("url","name","salary","province","city","exp","edu","num","pubtime","cname","ctype","ctrade","cnum","cate1","cate2","welfare","detail")
      .map(col=>StructField(col,DataTypes.StringType)))
    val data = spark.read.option("header","false").schema(schema).csv("result/cleaned/*")
    println(data.count())
//      转换工资为整数
//       .withColumn("salary",col("salary").cast(DataTypes.IntegerType))
//      .withColumn("num",col("num").cast(DataTypes.IntegerType))
//      .as[Job](jobEncoder)
//    data.filter(job=>job.cate1=="后端开发" && job.cate2=="其他").select("name","detail").write.csv("result/后端开发其他/")
//    福利
//    val r = data.select("welfare").rdd.flatMap(row=>{
//      val s = row.getAs[String]("welfare")
//      if(s==null)
//        Array[String]()
//      else
//      s.split("\\|")
//    }).map(f=>(f,1)).reduceByKey(_+_).collect().foreach(println)

//    isnull(
//
//    data.groupBy("num").agg(
//      count("num").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
//
//
//    println("----")
//    data.groupBy("ctype").agg(
//      count("ctype").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
//    println("----")
//    data.groupBy("cate2").agg(
//      count("cate2").alias("c")
//    ).orderBy(desc("c")).limit(100)
//      .collect()
//      .foreach(println)
//    data.groupBy("province").agg(
//      count("province").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
//
////    统计各个省的数量
//    data.groupBy("city").agg(
//      count("city").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
//
//    data.groupBy("exp").agg(
//      count("exp").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
//
//    data.groupBy("edu").agg(
//      count("edu").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
//
////    data.groupBy("ctrade").agg(
////      count("ctrade").alias("c")
////    ).orderBy(desc("c"))
////      .collect()
////      .foreach(println)
//
//    data.groupBy("cnum").agg(
//      count("cnum").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
//    data.groupBy("pubtime").agg(
//      count("pubtime").alias("c")
//    ).orderBy(desc("c"))
//      .collect()
//      .foreach(println)
  }
}
